package net.wlm.jtp.dwd.log.job;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.wlm.jtp.dwd.log.util.JdbcUtil;
import net.wlm.jtp.dwd.log.util.KafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class JtpLogClickHouse {
    public static void main(String[] args) throws Exception{
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(1);
        // 读取数据
        DataStream<String> pageStream = KafkaUtil.consumerKafka(env, "dwd-traffic-page-log");
        DataStream<String> startStream = KafkaUtil.consumerKafka(env, "dwd-traffic-start-log");
        DataStream<String> errStream = KafkaUtil.consumerKafka(env, "dwd-traffic-error-log");
        // pageStream.print("kafka");
        // 页面浏览日志数据写入
        savePageLog(pageStream);
        saveStartLog(startStream);
        // 脏数据写入
        saveErrLog(errStream);
        // 执行
        env.execute();
    }

    private static void saveErrLog(DataStream<String> errStream) {
        SingleOutputStreamOperator<String> errObject = errStream.map(new RichMapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                // 解析JSON
                JSONObject object = JSON.parseObject(s);
                JSONObject commonObject = object.getJSONObject("common");
                JSONObject errObject = object.getJSONObject("err");
                // 提取字段值
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("mid", commonObject.getString("mid"));
                jsonObject.put("area_code", commonObject.getString("ar"));
                jsonObject.put("brand", commonObject.getString("ba"));
                jsonObject.put("channel", commonObject.getString("ch"));
                jsonObject.put("is_new", commonObject.getString("is_new"));
                jsonObject.put("model", commonObject.getString("md"));
                jsonObject.put("os", commonObject.getString("os"));
                jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString());
                jsonObject.put("version_code", commonObject.getString("vc"));
                jsonObject.put("msg", errObject.getString("msg"));
                jsonObject.put("error_code", errObject.getIntValue("error_code"));
                jsonObject.put("ts", object.getLong("ts"));
                // 返回
                return jsonObject.toJSONString();
            }
        });
        // 字段名称
        String[] columns = new String[]{
                "mid", "area_code","brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "msg", "error_code","ts"
        };
        // 插入语句
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_app_err_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    msg, error_code, ts\n" +
                ") VALUES (\n" +
                "     ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                " )" ;
        // 插入数据
        JdbcUtil.sinkToClickhouseUpsertJson(errObject,columns, upsertSql);
    }
    private static void saveStartLog(DataStream<String> startStream) {
        SingleOutputStreamOperator<String> startObject = startStream.map(new RichMapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                // 解析JSON
                JSONObject object = JSON.parseObject(s);
                JSONObject commonObject = object.getJSONObject("common");
                JSONObject startObject = object.getJSONObject("start");
                // 提取字段值
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("mid", commonObject.getString("mid"));
                jsonObject.put("area_code", commonObject.getString("ar"));
                jsonObject.put("brand", commonObject.getString("ba"));
                jsonObject.put("channel", commonObject.getString("ch"));
                jsonObject.put("is_new", commonObject.getString("is_new"));
                jsonObject.put("model", commonObject.getString("md"));
                jsonObject.put("os", commonObject.getString("os"));
                jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString());
                jsonObject.put("version_code", commonObject.getString("vc"));
                jsonObject.put("entry", startObject.getString("entry"));
                jsonObject.put("loading_time", startObject.getLong("loading_time"));
                jsonObject.put("open_ad_id", startObject.getLong("open_ad_id"));
                jsonObject.put("open_ad_ms", startObject.getLong("open_ad_ms"));
                jsonObject.put("open_ad_skip_ms", startObject.getLong("open_ad_skip_ms"));
                jsonObject.put("ts", object.getLong("ts"));
                // 返回
                return jsonObject.toJSONString();
            }
        });
        // 字段名称
        String[] columns = new String[]{
                "mid", "area_code","brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "entry", "loading_time", "open_ad_id", "open_ad_ms", "open_ad_skip_ms",  "ts"
        };
        // 插入语句
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_app_start_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    entry, loading_time, open_ad_id, open_ad_ms, open_ad_skip_ms,\n" +
                "    ts\n" +
                ") VALUES (\n" +
                "     ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                " )" ;
        // 插入数据
        JdbcUtil.sinkToClickhouseUpsertJson(startObject,columns, upsertSql);
    }
    private static void savePageLog(DataStream<String> pageStream) {
        // 数据转换
        SingleOutputStreamOperator<String> jsonStream = pageStream.map(new RichMapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                // 解析json
                JSONObject object = JSON.parseObject(s);
                JSONObject commonObject = object.getJSONObject("common");
                JSONObject pageObject = object.getJSONObject("page");
                // 提取字段值
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("mid", commonObject.getString("mid"));
                jsonObject.put("area_code", commonObject.getString("ar"));
                jsonObject.put("brand", commonObject.getString("ba"));
                jsonObject.put("channel", commonObject.getString("ch"));
                jsonObject.put("is_new", commonObject.getString("is_new"));
                jsonObject.put("model", commonObject.getString("md"));
                jsonObject.put("os", commonObject.getString("os"));
                jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString());
                jsonObject.put("version_code", commonObject.getString("vc"));
                jsonObject.put("page_id", pageObject.getOrDefault("page_id", ""));
                jsonObject.put("last_page_id", pageObject.getOrDefault("last_page_id", ""));
                jsonObject.put("during_time", pageObject.getLong("during_time"));
                jsonObject.put("sourceType", pageObject.getOrDefault("sourceType", ""));
                jsonObject.put("item_type", pageObject.getOrDefault("item_type", ""));
                jsonObject.put("item", pageObject.getOrDefault("item", ""));
                jsonObject.put("ts", object.getLong("ts"));
                // 返回
                return jsonObject.toJSONString();
            }
        });
        // 字段名
        String[] columns =  new String[]{"mid","area_code","brand","channel",
                "is_new","model","os","uid","version_code","page_id",
                "last_page_id","during_time","sourceType","item_type","item","ts"
        };
        // 插入sql
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_page_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    page_id, last_page_id, during_time, sourceType, item_type, item,\n" +
                "    ts\n" +
                ") VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? )";
        // 写入clickhouse
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream,columns,
                upsertSql);
    }
}